package mqtttool

import (
	"errors"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"local.public/innerpkg/logger"
	"log"
	"os"
)

var mqttCli *mqtt.Client

func InitMqttConn(strBrokerAddr, strClientId string, f mqtt.MessageHandler) error {
	tmpMqttClient, err := NewMqttConn(strBrokerAddr, strClientId, f)
	if err != nil {
		return err
	}
	mqttCli = tmpMqttClient

	logger.Info("初始化mqtt连接完成")
	return nil
}

func NewMqttConn(strBrokerAddr, strClientId string, defaultFunc mqtt.MessageHandler) (*mqtt.Client, error) {
	mqtt.ERROR = log.New(os.Stdout, "", 0)

	opts := mqtt.NewClientOptions().AddBroker(strBrokerAddr).SetClientID(strClientId)
	//opts.SetKeepAlive(2 * time.Second)
	opts.SetDefaultPublishHandler(defaultFunc)
	//opts.SetPingTimeout(1 * time.Second)

	// 启动一个链接
	c := mqtt.NewClient(opts)
	if token := c.Connect(); token.Wait() && token.Error() != nil {
		logger.Error("mqtt连接发生错误")
		panic(token.Error())
	}

	logger.Debug("mqtt连接成功," + strBrokerAddr + ", clientid:" + strClientId)

	// 订阅worker的状态上报
	topic := "workeradmin/report/#"
	if token := c.Subscribe(topic, 0, nil); token.Wait() && token.Error() != nil {
		fmt.Println(topic + "订购失败")
		fmt.Println(token.Error())
		os.Exit(1)
	}
	logger.Debug(topic + "订购成功")
	//// 订阅worker的状态上报
	//if token := c.Subscribe("worker/report/#", 0, nil); token.Wait() && token.Error() != nil {
	//	fmt.Println("worker/report/#" + "订购失败")
	//	fmt.Println(token.Error())
	//	os.Exit(1)
	//}
	//logger.Debug("worker/report/#" + " 订购成功")
	return &c, nil
}

func AddMqttSubscribe(subjectPattern string) error {
	// 订阅worker的状态上报
	if token := (*mqttCli).Subscribe(subjectPattern, 0, nil); token.Wait() && token.Error() != nil {
		fmt.Println(subjectPattern + " 订购失败")
		fmt.Println(token.Error())
		return errors.New(subjectPattern + " 订购失败")
	}
	logger.Debug(subjectPattern + " 订购成功")
	//logger.Debug(utils.MqttSubscribeWorkerReportPattern + "订购成功")

	//// 订阅worker的状态上报
	//if token := mqttCli.Subscribe(utils.MqttSubscribeWorkerReportPattern, 0, nil); token.Wait() && token.Error() != nil {
	//	fmt.Println(utils.MqttSubscribeWorkerReportPattern + "订购失败")
	//	fmt.Println(token.Error())
	//	os.Exit(1)
	//}
	//logger.Debug(utils.MqttSubscribeWorkerReportPattern + "订购成功")
	//if _, err := natsClient.Subscribe(subj, cb); err != nil {
	//	logger.Error(subj+"订购失败", err.Error())
	//	return err
	//}
	return nil
}

func CloseMqttConn() {
	if mqttCli != nil {
		(*mqttCli).Disconnect(250)
	}
}

func MqttSend(topic string, payload string) {
	if mqttCli != nil {
		if !(*mqttCli).IsConnected() {
			logger.Debug("没有mqtt连接")
			return
		}
		token := (*mqttCli).Publish(topic, 0, false, payload)
		if wResult := token.Wait(); !wResult {
			logger.Debug("wResult is false ")
		}
	} else {
		logger.Debug("mqttCli is nil " + "MqttSend 失败")
	}
}
